import com.rabbitmq.client.Channel
import com.rabbitmq.client.Connection
import com.rabbitmq.client.ConnectionFactory
import com.rabbitmq.client.Envelope
import com.rabbitmq.client.QueueingConsumer

/**
 * Testing rabbitmq with Producer and Consumer
 * @author: linyu
 * @version: 3/22/16
 */


def Sending() {
    String DEFAULT_TOPIC = "one.two.three.four";

    String uri = (args.length > 0) ? args[0] : "amqp://localhost";
    String topic = (args.length > 1) ? args[1] : DEFAULT_TOPIC;
    String exchange = (args.length > 2) ? args[2] : null;
    String message = (args.length > 3) ? args[3] :
            "the time is " + new java.util.Date().toString();

    ConnectionFactory cfconn = new ConnectionFactory();
    cfconn.setUri(uri);
    Connection conn = cfconn.newConnection();

    Channel ch = conn.createChannel();

    if (exchange == null) {
        exchange = "amq.topic";
    } else {
        ch.exchangeDeclare(exchange, "topic");
    }

    for(int i=0;i<10;i++) {
        ch.basicPublish(exchange, topic, null, message.getBytes());
        System.out.println("Sending to exchange " + exchange + ", topic " + topic);

    }
    ch.close();
    conn.close();

}

def Recieving() {
    String DEFAULT_TOPIC = "one.two.three.four";

    String uri = (args.length > 0) ? args[0] : "amqp://localhost";
    String topicPattern = (args.length > 1) ? args[1] : DEFAULT_TOPIC;
    String exchange = (args.length > 2) ? args[2] : null;
    String queue = (args.length > 3) ? args[3] : null;

    ConnectionFactory cfconn = new ConnectionFactory();
    cfconn.setUri(uri);
    Connection conn = cfconn.newConnection();

    final Channel channel = conn.createChannel();

    exchange = "amq.topic";
//    if (exchange == null) {
//
//    } else {
//        channel.exchangeDeclare(exchange, "topic");
//    }

    queue = channel.queueDeclare().getQueue();

//    if (queue == null) {
//
//    } else {
//        channel.queueDeclare(queue, false, false, false, null);
//    }

    println channel.queueBind(queue, exchange, "one.*");

    System.out.println("Listening to exchange " + exchange + ", pattern " + DEFAULT_TOPIC +
            " from queue " + queue);

    QueueingConsumer consumer = new QueueingConsumer(channel);
    channel.basicConsume(queue, consumer);
    while (true) {
        println 1
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        println 2
        Envelope envelope = delivery.getEnvelope();
        System.out.println(envelope.getRoutingKey() + ": " + new String(delivery.getBody(), "UTF-8"));
        channel.basicAck(envelope.getDeliveryTag(), false);
    }
}

Sending()
Recieving()
